package com.datamining.online

import java.util.{Calendar, Properties}

import org.apache.spark.sql.Row;
import org.apache.spark.SparkConf
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{SQLContext, SaveMode, SparkSession}

/**
  * Created by Administrator on 2017/5/9.
  */
/**
  * HiveTestByDay
  *
  * @author Administrator kevin
  * @create 2017-05-09 17:47
  */
object HiveTestByDay {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf();
    sparkConf.setMaster("yarn-cluster");
    sparkConf.setAppName("my_test");

    val jdbc_url = "jdbc:mysql://192.168.20.234:3306/yamibuy_statistics"
    val table_name = "statistics_item_repurchase_rate"
    val properties = new Properties()
    properties.setProperty("user", "root")
    properties.setProperty("password", "root")
    properties.setProperty("driver", "com.mysql.jdbc.Driver")

    // 创建sparkSession
    val sparkSession = SparkSession.builder().appName("kevin-test").config(conf = sparkConf).enableHiveSupport().getOrCreate();

    import sparkSession.implicits._
    import sparkSession.sql

//    sql("use yamibuy_master")
    val sqlDataFrameOrderInfo = sql("SELECT order_id,order_sn,order_status,pay_status,user_id,add_time,from_unixtime(add_time,'yyyy-MM-dd') AS add_date FROM yamibuy_master.xysc_order_info WHERE pay_status > 0")

    val sqlDataFrameOrderGoods = sql("SELECT order_id,goods_id,goods_number FROM yamibuy_master.xysc_order_goods WHERE vendor_id = 0")

    val order_item_join = sqlDataFrameOrderInfo.join(sqlDataFrameOrderGoods, "order_id")

    val user_item_add_time_desc = order_item_join.orderBy(order_item_join("add_time").desc)
    val tmpVeiw = user_item_add_time_desc.createOrReplaceTempView("user_item_add_time_desc")

    val sqlDataFrameGoodsGoodsNumber = sql("SELECT goods_id, add_date, SUM(goods_number) AS goods_number FROM user_item_add_time_desc WHERE goods_id > 0 GROUP BY goods_id, add_date ORDER BY goods_id ASC,add_date DESC")
//    sqlDataFrameGoodsGoodsNumber.show(10)

    val sqlDataFrameGoodsGoodsNumberRdd = sqlDataFrameGoodsGoodsNumber.rdd

    val ttt_rdd = sqlDataFrameGoodsGoodsNumberRdd.map(x => (x(0), (x(1), x(2), 0.toDouble)))

    // add_time倒序
    val ttt_rdd_sort = ttt_rdd.groupByKey().sortBy(x => x._2.map(y => y._1.toString), false)
//    ttt_rdd_sort.take(10)

    // 每个商品保留最近100条记录
    val ttt_rdd_sort_sub = ttt_rdd_sort.map(x => (x._1, x._2.splitAt(100)._1))

    val item_repurchasing = ttt_rdd_sort_sub.map(x => {
      val item_time_quantity_array = x._2.toArray
      val length = item_time_quantity_array.size
      for (i <- 0 to length - 2) {

        val item_1 = item_time_quantity_array.apply(i)
        val item_2 = item_time_quantity_array.apply(i + 1)

        val time_1 = item_1._1
        val time_2 = item_2._1
        val quantity = item_2._2

        //        // 按秒
        //        val Tn = (time_1.toString.toDouble - time_2.toString.toDouble) / quantity.toString.toDouble
        // 按天
        val Tn = compareDate(time_1.toString, time_2.toString).toDouble / quantity.toString.toDouble

        item_time_quantity_array.update(i, (item_1._1, item_1._2, Tn))
      }
      (x._1, item_time_quantity_array)
    })

    val item_repurchasing_flat_value = item_repurchasing.flatMapValues(x => x)
    val item_repurchasing_flat_value_lll = item_repurchasing_flat_value.map(x => (x._1, x._2._1, x._2._2, x._2._3))

    item_repurchasing_flat_value_lll.take(10)
    val row_rdd = item_repurchasing_flat_value_lll.map(x => Row(Integer.valueOf(x._1.toString), String.valueOf(x._2.toString), Integer.valueOf(x._3.toString), x._4.toDouble))

    val schema = StructType(Array(StructField("goods_id", DataTypes.IntegerType, true), StructField("date", DataTypes.StringType, true), StructField("quantity", DataTypes.IntegerType, true), StructField("back_cycle", DataTypes.DoubleType, true)))
    sparkSession.createDataFrame(row_rdd, schema).write.mode(SaveMode.Overwrite).jdbc(jdbc_url, table_name, properties)
    sparkSession.close()
  }

  def compareDate(dateStr1: Any, dateStr2: Any): Int = {
    val dateStra = dateStr1.toString
    val calendar1: Calendar = Calendar.getInstance();
    calendar1.set(dateStra.split("-").apply(0).toInt, dateStra.split("-").apply(1).toInt, dateStra.split("-").apply(2).toInt)

    val dateStrb = dateStr2.toString
    val calendar2: Calendar = Calendar.getInstance();
    calendar2.set(dateStrb.split("-").apply(0).toInt, dateStrb.split("-").apply(1).toInt, dateStrb.split("-").apply(2).toInt)

    return ((calendar1.getTimeInMillis - calendar2.getTimeInMillis) / (1000 * 3600 * 24)).abs.toInt
  }
}
